package spark_core;

import java.util.ArrayList;
import java.util.Arrays;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.util.AccumulatorV2;
import org.apache.spark.util.LongAccumulator;
import scala.reflect.ClassManifestFactory;

/**
 * @author shihb
 * @date 2020/1/9 11:43
 * 累加器:分布式只写共享变量
 * 广播变量:分布式只读共享变量
 */
public class ShareData {

  public static void main(String[] args) {
    //local模式,创建SparkConf对象设定spark的部署环境
    SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("mark rdd");
    //创建spark上下文对象（这边是java上下文）
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);


    JavaRDD<Integer> rdd = jsc.parallelize(Arrays.asList(1,2,3,4,5,6),2);
    int multiple=2;
    //使用广播变量减少数据的传输(传给excute),使性能得到提高,是个调优的策略
    Broadcast<Integer> broadcast = jsc.sc()
        .broadcast(multiple, ClassManifestFactory.fromClass(Integer.class));

    //使用累加器来共享变量，来累加数据
    LongAccumulator longAccumulator = jsc.sc().longAccumulator();
    rdd.foreach(i->longAccumulator.add(i*broadcast.value()));

    System.out.println("sum="+longAccumulator.value());
    //停止
    jsc.stop();

  }
}

/**
 * 自定义累加器
 * 如果要使用,先new一个myAccumulator
 * 然后注册累加器jsc.sc().register(spark_core.myAccumulator);
 */
class myAccumulator extends AccumulatorV2<String, ArrayList<String>>{

  private ArrayList arrayList = new ArrayList();
  /**
   * 当前的累加器是否为初始状态
   * @return true为初始
   */
  @Override
  public boolean isZero() {

    return arrayList.isEmpty();
  }

  /**
   * 复制累加器对象
   * @return 累加器
   */
  @Override
  public AccumulatorV2 copy() {
    return new myAccumulator();
  }

  /**
   * 重置累加器
   */
  @Override
  public void reset() {
    arrayList.clear();
  }

  /**
   * 增加输入
   * @param s 输入
   */
  @Override
  public void add(String s) {
    arrayList.add(s);
  }

  /**
   * 获取累加器的结果
   * @return
   */
  @Override
  public ArrayList<String> value() {
    return arrayList;
  }

  /**
   * 合并累加器
   * @param accumulatorV2 累加器
   */
  @Override
  public void merge(AccumulatorV2<String, ArrayList<String>> accumulatorV2) {
    arrayList.addAll(accumulatorV2.value());
  }
}